package cn.yi.fast.mq.rocketmq.service.impl;

import cn.yi.fast.mq.rocketmq.properties.RocketMQProperties;
import cn.yi.fast.mq.rocketmq.service.RocketMQPullService;
import com.alibaba.rocketmq.client.consumer.MQPullConsumer;
import com.alibaba.rocketmq.client.consumer.MQPullConsumerScheduleService;
import com.alibaba.rocketmq.client.consumer.PullResult;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
@Slf4j
public class RocketMQPullServiceImpl implements RocketMQPullService {
    @Override
    public void pullMessageQueue(RocketMQProperties properties, MessageInterface messageInterface) {
        //消费组
        final MQPullConsumerScheduleService scheduleService = new MQPullConsumerScheduleService(properties.getGroupName());
        //MQ NameService地址
        scheduleService.getDefaultMQPullConsumer().setNamesrvAddr(properties.getNamesrvAddr());
        //负载均衡模式
        scheduleService.setMessageModel(MessageModel.CLUSTERING);
        //需要处理的消息topic
        scheduleService.registerPullTaskCallback("TopicTest1", (mq, context) -> {
            MQPullConsumer consumer = context.getPullConsumer();
            try {
                long offset = consumer.fetchConsumeOffset(mq, false);
                if (offset < 0)
                    offset = 0;
                PullResult pullResult = consumer.pull(mq, "*", offset, 32);
                log.info("%s%n", offset + "\t" + mq + "\t" + pullResult);
                switch (pullResult.getPullStatus()) {
                    case FOUND:
                        List<MessageExt> messageExts = pullResult.getMsgFoundList();
                        Object processResult =  messageInterface.process(messageExts);
                        log.info("process current messages result = {}",processResult);
                        consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());
                        break;
                    case NO_MATCHED_MSG:
                        break;
                    case NO_NEW_MSG:
                    case OFFSET_ILLEGAL:
                        break;
                    default:
                        break;
                }
                context.setPullNextDelayTimeMillis(100);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}
